package kafka

import (
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
	"go_knowledge/mq/kafka/event"
	"go_knowledge/mq/kafka/event/kafka/producer"
)

// 根据事件自动推送到不同的TOPIC
type autoTopicPusher struct {
	addr   []string
	logger *logrus.Logger
}

func NewAutoTopicPusher(addr []string) *autoTopicPusher {
	return &autoTopicPusher{addr: addr}
}

func (atp *autoTopicPusher) SetLogger(logger *logrus.Logger) {
	atp.logger = logger
}

func (atp *autoTopicPusher) SyncPush(event event.PbEvent, blockKey string) error {
	queueName, err := event.MakeTopic(event)
	if nil != err {
		return errors.Wrap(err, "make topic by event failed")
	}

	client := producer.NewDefaultSyncProducerClient(atp.addr, queueName)
	if nil != atp.logger {
		client.SetLogger(atp.logger)
	}

	return client.PushPbEvent(event, blockKey)
}

func (atp *autoTopicPusher) Push(event event.PbEvent, blockKey string) {
	go func() {
		if err := atp.SyncPush(event, blockKey); nil != err {
			atp.logger.Errorf("push event failed: %+v", err)
		}
	}()
}
